package cn.doitedu.dw_etl

import ch.hsr.geohash.GeoHash
import cn.doitedu.dw_beans.ApplogBean
import org.apache.commons.io.FileUtils
import org.apache.commons.lang3.time.{DateFormatUtils, DateUtils}
import org.apache.hadoop.conf.Configuration
import org.apache.hadoop.fs.{FSDataInputStream, FileStatus, FileSystem, Path}
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.{DataFrame, Dataset, Row, SaveMode, SparkSession}
import org.lionsoul.ip2region.{DataBlock, DbConfig, DbSearcher}

import java.io.{File, FileInputStream}
import java.net.URI

/**
 * @author 涛哥
 * @nick_name "deep as the sea"
 * @contact qq:657270652 wx:doit_edu
 * @site www.doitedu.cn
 * @date 2021-12-13
 * @desc ods数据集成维度信息（地理位置，guid）--> 维度退化
 */
object ApplogOds2DwdIntegrate {
  def main(args: Array[String]): Unit = {

    if(args.size<1){
      println(
        """
          |
          |params not enough!
          |usage:
          |  args(0): 要计算的分区日期
          |
          |""".stripMargin)
      sys.exit(1)
    }

    val dt = args(0)

    val spark: SparkSession = SparkSession.builder()
      .appName("ods数据集成维度信息")
      //.master("local")
      .enableHiveSupport()
      .getOrCreate()

    import spark.implicits._

    // 读取清洗过滤[cn.doitedu.dw_etl.ApplogOds2DwdPreprocess]的处理结果
    val ds: Dataset[ApplogBean] = spark.read.parquet(s"hdfs://doit01:8020/tmp/tasktmp/sessionSplited/${dt}/").as[ApplogBean]

    // 集成地理位置信息（将日志中的gps坐标转成geohash码，去维表：geohash地理信息维表  查询省市区）
    // 如果日志中没有gps坐标，或者上面的维表查不到，则用ip地址去查

    // 加载hive中的geohash维表
    val geoDataFrame: DataFrame = spark.read.table("dim.geo_area")

    // 将hive中读到的geohash维表，转成kv，收集到Driver端
    val geoMap = geoDataFrame.rdd
      // 把row转成 Option[ApplogBean]
      .map({
        case Row(province: String, city: String, district: String, geohash: String) => Some((geohash, (province, city, district)))
        case _ => None
      })
      // 过滤掉转失败的
      .filter(_.isDefined)
      // 将 Option[ApplogBean] 取到里面的 ApplogBean
      .map(_.get)
      // 将dataframe收集成Driver端的一个单机集合对象
      .collectAsMap()

    // 广播这个geohash字典单机集合
    val bc1 = spark.sparkContext.broadcast(geoMap)

    // 用单机文件流api，加载ip2region的字典库文件成： 字节数组
    /*val bytes: Array[Byte] = FileUtils.readFileToByteArray(new File("ip2region.db"))*/
    val conf = new Configuration()
    conf.set("fs.hdfs.impl.disable.cache","true")
    val fs: FileSystem = FileSystem.get(new URI("hdfs://doit01:8020/"), conf)

    val dbPath = new Path("/data/ip2region/ip2region.db")
    val in: FSDataInputStream = fs.open(dbPath)
    val statuse: FileStatus = fs.listStatus(dbPath)(0)

    val bytes = new Array[Byte](statuse.getLen.toInt)
    in.readFully(bytes)

    in.close()
    fs.close()

    // 将这个字典文件的字节数组对象广播出去
    val bc2 = spark.sparkContext.broadcast(bytes)

    /**
     * 正式的处理逻辑：
     *  对清洗过滤后的日志数据，逐条处理
     */
    val areaIntegrated: Dataset[ApplogBean] = ds.mapPartitions(iter => {
      // 从广播变量中取到geohash字典HashMap集合
      val geoDict = bc1.value

      // 从广播变量中取到ip2region的字典文件字节数组对象
      val ip2RegionBytes: Array[Byte] = bc2.value

      // 构造一个ip2region的查询工具
      val config = new DbConfig()
      val searcher = new DbSearcher(config, ip2RegionBytes)

      // 开始逐条处理清洗过滤后的行为日志
      iter.map(bean => {
        // 定义3个临时变量，用来记录查询到的省市区
        var province: String = "未知省"
        var city: String = "未知市"
        var district: String = "未知区"


        // 从日志bean中取出经纬度
        val lng: Double = bean.longitude
        val lat: Double = bean.latitude

        // 判断经纬度是否合法
        if (lng != null && lat != null && lat <= 65 && lat >= -25 && lng >= 0 && lng <= 180) {
          // 将经纬度转成geohash编码
          val geoCode: String = GeoHash.geoHashStringWithCharacterPrecision(lat, lng, 6)

          // 去geohash维表HashMap中查询省市区
          val areas: Option[(String, String, String)] = geoDict.get(geoCode)
          // 如果查询结果有值
          if (areas.isDefined) {
            // 将查询结果中的省市区赋值给前面定义的3个临时变量
            province = areas.get._1
            city = areas.get._2
            district = areas.get._3
          }
        }

        // 如果通过gps维表没有查到省市区,用ip地址去查询
        if ("未知省".equals(province)) {
          // 利用ip2region的查询工具，查询本条日志bean中ip
          val block: DataBlock = searcher.memorySearch(bean.ip)

          // 995|中国|0|上海|上海市|电信|129764
          // 切割查询结果
          val arr: Array[String] = block.toString.split("\\|")
          // 判断查询结果切割后的字段个数是否为7个，以及省字段是否不等于0
          if (arr.size == 7 && !arr(3).equals("0")) {
            // 从查询结果中取脚标3字段赋值给临时变量province
            province = arr(3)
            city = arr(4)
          }
        }

        // 将查询到的结果，赋值给bean对象的成员
        bean.province = province
        bean.city = city
        bean.district = district

        // 返回添加了省市区信息的bean
        bean
      })
    })

    areaIntegrated.write/*.mode(SaveMode.Overwrite)*/.parquet(s"hdfs://doit01:8020/tmp/tasktmp/areaAdded/${dt}/")

    spark.close()

  }

}
